/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl.disk;

import java.io.IOException;
import java.util.WeakHashMap;


/**
 * Disk storage keeps data records on disk and references in memory. 
 * Memory consumption is proportional to number of entries and does not depend on size of the 
 * entries.
 * Call compact() periodically.It cleans unused disk space.
 * Limits: number of records <= 2^31, size of the file <= 2^40
 * Notes: It does not cache or compress data, use external systems to 
 * reduce IO operations.
 * The storage supports READ COMMITTED isolation level.
 * Use Transaction.tryLock in order to enforce higher isolation levels.
 * @author Alex Antonau
 * */
public class DiskStorageImpl extends AbstractDiskStorage implements DiskStorage {

	private final WeakHashMap<DiskTransactionImpl, Object> transactions;
	
	public DiskStorageImpl(String workPath) throws IOException {
		super(workPath);
		transactions = new WeakHashMap<DiskTransactionImpl, Object>();
	}

	public DiskTransactionImpl beginTransaction() throws IOException {
		checkOpen();
		DiskTransactionImpl t = new DiskTransactionImpl(this);
		synchronized(transactions){
			transactions.put(t, null);
		}
		return t;
	}
	
	protected void completeDiskTransaction(DiskTransactionImpl t){
		synchronized(transactions){
			transactions.remove(t);
		}
	}
	protected void failUnflushedTransactions(IOException ex){
		synchronized(transactions){
			for(DiskTransactionImpl t:transactions.keySet()){
				if(!t.isFlushed()) t.setFailure(ex);
			}
		}
	}
	protected DiskTransactionImpl tryLock(long[] addresses, DiskTransactionImpl current){
		synchronized(transactions){
			for(DiskTransactionImpl t:transactions.keySet()){
				if(t.isLocked(addresses)) return t;
			}
			current.setLock(addresses);
		}
		return current;
	}
	
	protected void write(DataEntry entry, byte[] data, int offset) throws IOException{
		try{
			dataLog.writeData(entry, data, offset);
		}catch (IOException e) {
			failUnflushedTransactions(e);
			throw e;
		} 
	}

	protected void flushCurrentLog() throws IOException{
		try{
			dataLog.flush();
		}catch (IOException e) {
			failUnflushedTransactions(e);			
			throw e;
		} 
	}

	protected boolean hasLogLocked(int log) {
		synchronized (transactions) {
			for(DiskTransactionImpl t : transactions.keySet()){
				if(t.hasLogLocked(log)) return true;
			}
		}
		return false;
	}

}
